package com.loveprogrammer.redismq.listener;

import com.alibaba.fastjson2.JSON;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.loveprogrammer.redismq.annotation.MQListener;
import com.loveprogrammer.redismq.config.RedisCache;
import com.loveprogrammer.redismq.config.SpringContextHelper;
import com.loveprogrammer.redismq.listener.handler.HandlerFactory;
import com.loveprogrammer.redismq.mq.MqUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName RedisKeyExpirationListener
 * @Description 键值失效监听类 - 当做
 * @Author admin
 * @Date 2023/10/9 17:07
 * @Version 1.0
 */
@Component
public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
    @Autowired
    private MqUtil mqUtil;

    @Autowired
    private HandlerFactory handlerFactory;

    @Autowired
    private RedisCache redisCache;

    private static final Logger logger = LoggerFactory.getLogger(RedisKeyExpirationListener.class);

    public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
        super(listenerContainer);
    }

    // 延时任务执行线程池
    private ThreadPoolExecutor executor = new ThreadPoolExecutor(
            3,
            10,
            0L,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(2048),
            new ThreadFactoryBuilder()
                    .setNameFormat("hyh-mq-pool-%d").build(),
            new ThreadPoolExecutor.AbortPolicy()
    );

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String expiredKey = message.toString();
        if (!expiredKey.startsWith("mq") || StringUtils.isBlank(expiredKey)) {
            return;
        }
        // 分布式锁-防止多次消费的情况
        String lock = redisCache.tryLock(expiredKey, 60);
        if (lock == null) {
            return;
        }
        // 获得值
        String value = mqUtil.getMessage(expiredKey);
        if (value == null) {
            value = "";
        }
        logger.info("获取到延时任务key {}", expiredKey);
        // 都是按照 mq:topic:tag:key
        String[] keys = expiredKey.split(":");
        if (keys.length < 4) {
            return;
        }
        String type = keys[1];
        String tag = keys[2];
        String key = keys[3];
        Class handler = handlerFactory.handlerMap.get(type);
        if (handler == null) {
            logger.warn("未获取到指定的任务对象，key {}", expiredKey);
            return;
        }
        String finalValue = value;
        String lockValue = lock;
        executor.execute(() -> {
            Boolean execute = false;
            try {
                Object bean = SpringContextHelper.getBean(handler);
                // 找到tag 遍历methods
                Method[] methods = handler.getMethods();
                for (Method method : methods) {
                    MQListener mqListener = method.getAnnotation(MQListener.class);
                    if (tag.equals(mqListener.tag())) {
                        Class<?> aClass = mqListener.messageClass();
                        String name = aClass.getName();
                        // 先处理基本类型
                        if("java.lang.String".equals(name)) {
                            method.invoke(bean, key, finalValue);
                        }else if("java.lang.Long".equals(name) ) {
                            Long object = Long.parseLong(finalValue);
                            method.invoke(bean, key, object);
                        }else if("java.lang.Integer".equals(name) ) {
                            Integer object = Integer.parseInt(finalValue);
                            method.invoke(bean, key, object);
                        }else if("java.lang.Short".equals(name) ) {
                            Short object = Short.parseShort(finalValue);
                            method.invoke(bean, key, object);
                        }else if("java.lang.Byte".equals(name) ) {
                            Byte object = Byte.parseByte(finalValue);
                            method.invoke(bean, key, object);
                        }else if("java.lang.Double".equals(name)) {
                            Double object = Double.parseDouble(finalValue);
                            method.invoke(bean, key, object);
                        }else if("java.lang.Float".equals(name)) {
                            Float object = Float.parseFloat(finalValue);
                            method.invoke(bean, key, object);
                        }
                        else{
                            Object object = JSON.parseObject(finalValue, aClass);
                            method.invoke(bean, key, object);
                        }
                        execute = true;
                        break;
                    }
                }
                if (!execute) {
                    logger.error("执行延时任务失败,60秒钟后重试");
                    mqUtil.sendMessage(expiredKey, finalValue, 60L, TimeUnit.SECONDS);
                }
            } catch (Exception e) {
                logger.error("执行延时任务失败,60秒钟后重试", e);
                mqUtil.sendMessage(expiredKey, finalValue, 60L, TimeUnit.SECONDS);
            }
            if (execute) {
                mqUtil.releaseMessage(expiredKey);
            }
            redisCache.unlock(expiredKey, lockValue);
        });


    }

    @PreDestroy
    public void destroy() {
        shutdownAsyncManager();
    }

    /**
     * 停止异步执行任务
     */
    private void shutdownAsyncManager() {
        try {
            logger.info("====关闭延时任务线程池====");
            executor.shutdown();
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }
}

